package com.jie.netty.listener;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.jie.constant.CommonConst;
import com.jie.constant.IMRabbitMqKey;
import com.jie.enums.IMCmdType;
import com.jie.enums.MessageStatus;
import com.jie.netty.model.IMRecvInfo;
import com.jie.netty.model.IMSendResult;
import com.jie.netty.processor.AbstractMessageProcessor;
import com.jie.netty.processor.ProcessorFactory;
import com.jie.pojo.SingleChat;
import com.jie.service.SingleChatService;
import com.jie.dto.PrivateMessageDTO;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

import static com.jie.constant.IMConstant.CHAT_RETRIES;

/**
 * @Author: 杰
 * @Project :netty-communication
 * @Description: 单聊消息监听处理
 * @DateTime: 2023/11/18 13:24
 **/
@Slf4j
@Component
public class PrivateMessageListener {

    @Autowired
    private SingleChatService singleChatService;

    private final RabbitTemplate rabbitTemplate;
    private final RedisTemplate redisTemplate;

    public PrivateMessageListener(RabbitTemplate rabbitTemplate, RedisTemplate redisTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        this.redisTemplate = redisTemplate;
    }

    //监听单聊信息发送成功队列
    @RabbitListener(queues = IMRabbitMqKey.IM_RESULT_PRIVATE_QUEUE)
    public void processPrivateChatMessages(Message message, Channel channel) throws IOException {
        try {
            IMRecvInfo imRecvInfo = JSON.parseObject(new String(message.getBody(), StandardCharsets.UTF_8), IMRecvInfo.class);
            Integer retriesCount = getRetriesCount(imRecvInfo);
            if (retriesCount >= CommonConst.MAX_RETRIES) {
                log.info("当前消息id:{}达到最高重试次数", imRecvInfo.getMessageId());
                // 失败确认，发送到死信队列让死信队列监听器进行人工处理
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
            } else {
                // netty发送消息
                boolean success = processMessage(imRecvInfo);
                if (success) {
                    // 消息处理成功，确认消息
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
                } else {
                    // 重试次数+1
                    incrementRetriesCount(imRecvInfo);
                    //前队列进行清除
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
                    // 重回队列
                    rabbitTemplate.convertAndSend(IMRabbitMqKey.IM_RESULT_PRIVATE_QUEUE, message);
                }
            }
        } catch (Exception e) {
            // 发生异常，发送到死信队列
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }
    }

    //监听单聊信息收到成功队列
    @RabbitListener(queues = IMRabbitMqKey.IM_RESULT_PRIVATE_RECEIVE_QUEUE)
    public void processResultPrivateChatMessages(Message message, Channel channel) throws IOException {
        try {
            IMSendResult<PrivateMessageDTO> result = JSON.parseObject(new String(message.getBody(), StandardCharsets.UTF_8), IMSendResult.class);
            //将成功投递消息进行状态的改变 有问题的消息进行ack
            if (result!=null&&result.getCode().equals(MessageStatus.UNSEND.code())){
                UpdateWrapper<SingleChat> updateWrapper = new UpdateWrapper<>();
                updateWrapper.lambda().eq(SingleChat::getId,result.getMessageId())
                        .eq(SingleChat::getState, MessageStatus.UNSEND.code())
                        .set(SingleChat::getState, MessageStatus.SENDED.code());
                singleChatService.update(updateWrapper);
                log.info("消息已读，消息id:{}，发送者:{},接收者:{},终端:{}",result.getMessageId(),result.getSender().getId(),result.getReceiver().getId(),result.getReceiver().getTerminal());
            }
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        }catch (Exception e){
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }
    }

    //监听信息发送失败
    @RabbitListener(queues = IMRabbitMqKey.DLX_CHAT_QUEUE)
    public void processDlxPrivateChatMessages(Message messages, Channel channel) throws IOException {
        //TODO 记录日志 让管理员进行手动补偿
        log.info("当前队列已超过最大次数也未能投递成功请等待管理员进行处理{}",messages);
        channel.basicAck(messages.getMessageProperties().getDeliveryTag(), true);
    }

    private Integer getRetriesCount(IMRecvInfo message) {
        // 获取消息的重试次数
        Integer retriesCount = (Integer) redisTemplate.opsForValue().get(CHAT_RETRIES + message.getMessageId());
        if (retriesCount == null) {
            retriesCount = 0;
        }
        return retriesCount;
    }
    //重试次数+1
    private void incrementRetriesCount(IMRecvInfo message) {
        String key = CHAT_RETRIES + message.getMessageId();
        redisTemplate.execute(new SessionCallback<Object>() {
            @Override
            public Object execute(RedisOperations operations) throws DataAccessException {
                operations.multi();
                operations.opsForValue().increment(key, 1);
                operations.expire(key, 60, TimeUnit.SECONDS);
                operations.exec();
                return null;
            }
        });
    }

    /**
     * 私聊发送消息
     * @param message
     * @return
     */
    private boolean processMessage(IMRecvInfo message) {
        if(message!=null){
            AbstractMessageProcessor processor = ProcessorFactory.createProcessor(IMCmdType.PRIVATE_MESSAGE);
            processor.process(message);
            return true;
        }
        return false;
    }
}
